package cn.jbooter.mq.autoconfigure.rocketmq.coms;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.jbooter.coms.exceptions.FrameworkException;


/**
 * rocketmq 生产者客户端
 * 
 * @author HeJian
 */
public class RocketTemplate {
	private static final Logger logger = LoggerFactory.getLogger(RocketTemplate.class);
	
	/** 队列选择器 **/
	private MessageQueueSelector intQueueSelector = new RocketSelectMessageQueueByIntSelector();
	/** rocketmq 默认的生产者 **/
	private DefaultMQProducer defaultMQProducer = null;
	/**事务消息生产者:(暂未支持)**/
	private TransactionMQProducer transactionMQProducer = null;
	
	public DefaultMQProducer getDefaultMQProducer() {
		return defaultMQProducer;
	}
	public void setDefaultMQProducer(DefaultMQProducer defaultMQProducer) {
		this.defaultMQProducer = defaultMQProducer;
	}

    /**
     * 使用defaultMQProducer同步发送消息
     * @param message  消息体
     * @throws FrameworkException 消息异常
     */
    public SendResult send(Message message) {
        try {
            SendResult rs = this.defaultMQProducer.send(message);
            logger.info("send rocketmq message ,messageId : {}", rs.getMsgId());
            return rs;
        } catch (Exception e) {
            throw new FrameworkException("defaultMQProducer消息发送失败，topic :" + message.getTopic() + ",e:" + e.getMessage());
        }
    }

    /**
     * 同步发送消息
     * @param message  消息体
     * @param queueSelector  int值,用于计算要选择的队列 eg: 1001
     * @throws FrameworkException 消息异常
     */
    public SendResult sendOrderly(Message message, int queueSelector){
        try {
            SendResult rs = this.defaultMQProducer.send(message,intQueueSelector , queueSelector);
            logger.info("send rocketmq message orderly ,messageId : {} ,queue id : {}", rs.getMsgId(),rs.getMessageQueue().getQueueId());
            return rs;
        } catch (Exception e) {
            throw new FrameworkException("defaultMQProducer顺序消息发送失败，topic :" + message.getTopic() + ",e:" + e.getMessage());
        }
    }

    /**
     * 异步发送消息
     * @param message msgObj
     * @param sendCallback 回调
     * @throws FrameworkException 消息异常
     */
    public void sendAsync(Message message, SendCallback sendCallback){
        try {
        	this.defaultMQProducer.send(message, sendCallback);
        	logger.info("send rocketmq message async(异步).");
        } catch (Exception e) {
            throw new FrameworkException("defaultMQProducer异步消息发送失败，topic :" + message.getTopic() + ",e:" + e.getMessage());
        }
    }
	

}